#include "base/base_inc.h"
#include "thread/epoll_thread.h"
#include "base/atomic.h"

namespace weilin {
	EpollThread::EpollThread(const uint32_t& socket_thread_num) :
	m_accept_io_service_sp(boost::make_shared<boost::asio::io_service>()), 
	m_accept_io_service_work_sp(new (std::nothrow)boost::asio::io_service::work(*m_accept_io_service_sp)),
	m_socket_thread_num(socket_thread_num),
	m_index(0)
	{
	}

	EpollThread::~EpollThread()
	{
		Stop(STOP_PEACE);
	}
	void EpollThread::Init()
	{
		for (uint32_t index = 0; index < m_socket_thread_num; index++)
		{
			io_service_shared_ptr io_service_sp(boost::make_shared<boost::asio::io_service>());
			io_service_work_shared_ptr io_service_work_sp(new (std::nothrow)boost::asio::io_service::work(*io_service_sp));
			m_socket_io_service_sp_vect.push_back(io_service_sp);
			m_socket_io_service_work_sp_vect.push_back(io_service_work_sp);
		}
	}

	void EpollThread::Start()
	{
		m_thread_group.create_thread(boost::bind(&EpollThread::Run, shared_from_this(), m_accept_io_service_sp));

		for (vector<io_service_shared_ptr>::iterator it = m_socket_io_service_sp_vect.begin(); it < m_socket_io_service_sp_vect.end(); it++)
		{
			m_thread_group.create_thread(boost::bind(&EpollThread::Run, shared_from_this(), *it));
		} 
	}

	void EpollThread::Stop(const StopMthod& stop_method)
	{
		if (stop_method == STOP_PEACE)
		{
			m_accept_io_service_work_sp.reset();
   		}
		else
		{
			m_accept_io_service_sp->stop();
			m_accept_io_service_sp.reset();
		}
	
		for (uint32_t index = 0; index < m_socket_io_service_sp_vect.size(); index++)
		{
			if (stop_method == STOP_PEACE)
			{
				m_socket_io_service_work_sp_vect[index].reset();
			}
			else
			{
				m_socket_io_service_sp_vect[index]->stop();
				m_socket_io_service_sp_vect[index].reset();
			}
		}
		m_thread_group.join_all();
	}


	boost::asio::io_service& EpollThread::GetSocketIoService()
	{
		io_service_shared_ptr io_service_sp = m_socket_io_service_sp_vect[AtomicIncrement<uint32_t>(&m_index) % m_socket_io_service_sp_vect.size()];
		return *io_service_sp;
	}

	boost::asio::io_service& EpollThread::GetAcceptIoService()
	{
		return *m_accept_io_service_sp;
	}


	void EpollThread::Run(const io_service_shared_ptr& io_service_sp)
	{
		io_service_sp->run();
	}
}
